package com.example.tms.realtime.app.ods;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.tms.realtime.utils.CreateEnvUtil;
import com.example.tms.realtime.utils.KafKaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.jline.utils.Log;

/**
 * Created with IntelliJ IDEA.
 * ClassName: OdsApp
 * Package: com.example.tms.realtime.app.ods
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-08-01
 * Time: 11:17
 */


//ods 数据采集
public class OdsApp {
    public static void main(String[] args) throws Exception {

        //TODO 1.获取流处理环境 指定检查点
        StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(args);
        env.setParallelism(4);

        //TODO 2.使用FlinkCDC从MySQL中读取数据 - 实时数据
        String dwdOption = "dwd";
        String dwdServerId = "6030";
        String dwdSourceName = "ods_app_dwd_source";
        mysqlToKafka(dwdOption, dwdServerId, dwdSourceName, env, args);


        // -----------------------------------------
        //实时数据处理完 之后 处理维度数据 过程是一样的 直接复制粘贴
        //对维度数据处理
        //TODO 2.使用FlinkCDC从MySQL中读取数据 - 实时数据
        String realtimeOption = "realtime_dim";
        String realtimeServerId = "6040";
        String realtimeSourceName = "ods_app_realtimeDim_source";
        mysqlToKafka(realtimeOption,realtimeServerId,realtimeSourceName, env, args);


        env.execute();

    }

    public static void mysqlToKafka(String option, String serverId, String sourceName, StreamExecutionEnvironment env, String[] args) {
        //TODO 2.使用FlinkCDC从MySQL中读取数据 - 实时数据
        MySqlSource<String> mySqlSource = CreateEnvUtil.getMySqlSource(option, serverId, args);

        //将MysqlSource封装成流
        SingleOutputStreamOperator<String> strDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), sourceName)
                .setParallelism(1) //source并行度设置为1 防止乱序出现
                .uid(option + sourceName);

        //TODO 3.简单的ETL
        //对数据进行清理 过滤算子 process也可以
        final SingleOutputStreamOperator<String> processDS = strDS.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<String> out) throws Exception {
                        //首先判断这个json字符串是不是完整的 标准的json字符串
                        try {
                            JSONObject jsonObject = JSON.parseObject(jsonStr);
                            //将字符串 转换为JSON
                            //如果是一个标准JSON那么就过滤
                            //获取到的数据 不是after不是null 且数据的操作熟悉不是delete 删除操作
                            if (jsonObject.getJSONObject("after") != null && !"d".equals(jsonObject.getString("op"))) {
                                //获取时间值
                                Long tsMs = jsonObject.getLong("ts_ms");
                                //在添加进去
                                jsonObject.put("ts", tsMs);
                                //在删掉原来的属性
                                jsonObject.remove("ts_ms"); //相当于该了名字
                                //在将json转换为字符串向下游传递
                                out.collect(jsonObject.toJSONString());
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                            //在转换的过程当中可能会发生异常
                            Log.error("从Flink-CDC得到的数据不是一个标准的json格式");
                        }
                    }
                }
        ).setParallelism(1); //全局是4 source是1 这个也设置为1 防止出现乱序

        //TODO 4.安装主键进行分组 避免出现乱序
        //存在问题是 上游并行度是1 但是 要往kafka四个分区里面写 也可能出现乱序的情况
        //先通过主键 进行分组 避免出现乱序的情况
        KeyedStream<String, String> keyDS = processDS.keyBy(
                new KeySelector<String, String>() {
                    @Override
                    public String getKey(String jsonStr) throws Exception {
                        //获取到json当中的主键
                        //先要将字符串转成JSON
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        return jsonObject.getJSONObject("after").getString("id");
                    }
                }
        );

        //TODO 5.将数据写到kafka
        //写入的主题名称 不能写死了
        keyDS.sinkTo(KafKaUtil.getKafkaSink("tms_ods",sourceName+"_transPre" ,args))
                .uid(option + "_ods_app_sink");

    }

}